package org.atguigu.gmall.realtime.app.dws;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.bean.KeywordBean;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.function.KwSplit;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dws
 * Author: ZengHaiFeng
 * CreateTime:2023/4/27 14:55:43
 * Description: Todo use flink sql API
 * Version：1.0
 */
public class Dws_01_DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new Dws_01_DwsTrafficSourceKeywordPageViewWindow()
                .init(40001,2,"Dws_01_DwsTrafficSourceKeywordPageViewWindow");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        // read page data from log dwd kafka topic
        tbe.executeSql("create table page (" +
                " page map<string,string> , " +
                " ts bigint, " +
                " et as to_timestamp_ltz(ts,3), " +
                " watermark for et as et - interval '3' second " +
                ")" +
                SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE,
                        "Dws_01_DwsTrafficSourceKeywordPageViewWindow"));
        /*
            Filter search record and extract the keyword
            There are two cases which search on the home page or search page.
            After searching the keyword, jump to result page all of their conditions.
            Hence, want to get search page through some conditions there are.
            The `last_page_id` is `home` or `search` and `item_type` is 'keyword' `item` is not null.
            And then, just query two fields that page's item and et.
            The item is needed to analysis of this time, and et field was used for a tumbling window.
         */
        Table kwTable = tbe.sqlQuery("select " +
                " page['item'] kw, " +
                " et " +
                "from page where " +
                " page['last_page_id'] in ('home','search') and " +
                " page['item_type']='keyword' and " +
                " page['item'] is not null ");
        tbe.createTemporaryView("kw_table",kwTable);

        // split keywords
        tbe.createTemporaryFunction("kw_split", KwSplit.class);
        // keyword is defined from kw_split that return the row.
        Table keywordTable = tbe.sqlQuery("select " +
                " keyword, " +
                " et " +
                " from kw_table " +
                " join lateral table(kw_split(kw)) on true");
        tbe.createTemporaryView("keyword_table",keywordTable);
        // aggregation with a tumbling window
        Table result = tbe.sqlQuery("select " +
                " date_format(window_start,'yyyy-MM-dd HH:mm:ss') stt, " +
                " date_format(window_end,'yyyy-MM-dd HH:mm:ss') edt, " +
                " keyword, " +
                " count(*) keyword_count, " +
                " unix_timestamp() * 1000 as ts " +
                " from " +
                " table(tumble(table keyword_table, descriptor(et), interval '5' second )) " +
                " group by window_start,window_end,keyword");
        // output the final result to clickhouse
        // convert the table environment to a stream
        SingleOutputStreamOperator<KeywordBean> stream = tbe
                .toRetractStream(result, KeywordBean.class)
                .filter(t -> t.f0)
                .map(t -> t.f1);
        stream.addSink(
                FlinkSinkUtil
                        .getClickHouseSink(
                                "dws_traffic_keyword_page_view_window",
                                KeywordBean.class));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
